package yartiss.framework.concurrency;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;

public class Producer<E>{

	private final BlockingQueue<QueueEntry<E>> queue;
	//private ExecutorService executor = Executors.newCachedThreadPool();
	private ExecutorService executor = Executors.newFixedThreadPool(3000);

	public Producer(BlockingQueue<QueueEntry<E>> queue){
		this.queue = queue;
	}


	public void produce(final Callable<E> callable){

		try{
			executor.submit(new Runnable() {

				@Override
				public void run() {
					try {
						if(callable==null)
							queue.put(new QueueEntry<E>(null));
						
						E result = callable.call();
						
						if(result == null){
							stop();
							return ;
						}
						queue.put(new QueueEntry<E>(result));
					} catch (Exception e) {	}
				}
			});
		}
		catch (RejectedExecutionException e) {}
	}

	public void awaitTermination(){
		executor.shutdown();
		while(!executor.isTerminated()){
			try {
				System.out.println(executor.awaitTermination(1, TimeUnit.DAYS));
			} catch (InterruptedException e) {}
		}
		System.out.println("Producer : awaitTermination finished");
	}


	public void stop(){
		executor.shutdownNow();
		System.out.println("Producer : stop finished");
	}
}
